package com.bigdata.spark.core.rdd.persist

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object RDD_Persist03 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("Persist")
    val sc = new SparkContext(sparkConf)

    val list = List("Hello Scala", "Hello Spark")

    val rdd = sc.makeRDD(list)
    val flatRDD = rdd.flatMap(_.split(" "))
    val mapRDD = flatRDD.map(word => {
      println("######################################")
      (word, 1)
    })

    // RDD中不存储数据
    // 如果一个RDD需要重复使用，那么需要从头再次执行获取数据
    // RDD对象可以重用的，但是数据无法重用
    // RDD对象的持久化操作不一定是为了重用
    // 在数据执行较长，或数据比较重要的场合也可以采用持久化操作

    // TODO - RDD 持久化
    // cache默认持久化的操作，只能将数据保存在内存中，如果想要保存到磁盘文件，需要给更改存储级别
    //mapRDD.cache()

    // 持久化操作必须在行动算子执行时完成的。
    mapRDD.persist(StorageLevel.DISK_ONLY)

    val reduceRDD = mapRDD.reduceByKey(_ + _)
    reduceRDD.collect().foreach(println)

    println("====================================================")

    val groupRDD = mapRDD.groupByKey()
    groupRDD.collect().foreach(println)

    sc.stop()
  }
}
